package com.crazymaker.springcloud.message.service.impl;

import com.crazymaker.push.platform.contract.constants.PushConstants;
import com.crazymaker.push.platform.contract.dto.MsgDTO;
import com.crazymaker.springcloud.common.exception.BusinessException;
import com.crazymaker.springcloud.common.util.JsonUtil;
import com.crazymaker.springcloud.message.dao.MsgDao;
import com.crazymaker.springcloud.message.dao.MsgStreamDao;
import com.crazymaker.springcloud.message.dao.po.MsgPO;
import com.crazymaker.springcloud.message.dao.po.MsgStreamPO;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;

/**
 * The class Mq message service.
 */
@Slf4j
@Service
public class MessagePushServiceImpl {

    @Resource
    private MsgDao msgDao;

    @Resource
    private MsgStreamDao msgStreamDao;


    @Resource
    MessageRelayService messageRelayService;


    @Value("${rocketmq.producer.topic}")
    private String topic;


    @Transactional(rollbackFor = {RuntimeException.class, Error.class})
    public void saveAndDispatchMessage(MsgDTO dto) {
        //topicKey 配置的topic
        String topicKey = topic + "-" + dto.getTargetType();
        //1 保持 msg
        MsgPO messagePO = msgDao.findByMsgId(dto.getMsgId());
        if (null == messagePO) {

            messagePO = new MsgPO();

            // 复制数据
            BeanUtils.copyProperties(dto, messagePO);
            //消息需要设置好ID
//        messagePO.setMsgId(null);
            messagePO.setCreateTime(new Date());
            messagePO.setMsgId(dto.getMsgId());

            messagePO = msgDao.saveAndFlush(messagePO);


//        String topicKey = PushConstants.getTopicKey(dto);
            //测试的时候，一定要发送type
            //ws的type为1
            String topicFlag = PushConstants.getTopicFlag(dto);
            System.out.println("messagePO = " + messagePO);

            String[] targetIds = dto.getTargetIds().split(",");
            for (int i = 0; i < targetIds.length; i++) {

                //1 保持 明细流水
                MsgStreamPO streamPO = new MsgStreamPO();
                BeanUtils.copyProperties(messagePO, streamPO);
                streamPO.setMsgId(null);
                streamPO.setSendTime(new Date());
                streamPO.setTargetId(targetIds[i]);
                streamPO.setReadStatus(PushConstants.PUSHED_BUT_UN_READ);
                streamPO.setSendStatus(PushConstants.PUSH_STARTING);
                streamPO.setMsgId(messagePO.getMsgId());
                streamPO.setAppKey(messagePO.getAppKey());

                msgStreamDao.saveAndFlush(streamPO);
                log.debug("streamPO = " + streamPO);

                sendToMq(topicKey, topicFlag, messagePO, streamPO);


            }
        } else {
            //保障幂等性


//        String topicKey = PushConstants.getTopicKey(dto);
            //测试的时候，一定要发送type
            //ws的type为1
            String topicFlag = PushConstants.getTopicFlag(dto);
            System.out.println("messagePO = " + messagePO);

            List<MsgStreamPO> streamList = msgStreamDao.findByMsgId(messagePO.getMsgId());
            for (MsgStreamPO streamPO : streamList) {


                sendToMq(topicKey, topicFlag, messagePO, streamPO);

            }


        }
    }

    /**
     * 按照 topic  flag 发送消息
     *
     * @param topicKey  topic
     * @param topicFlag flag
     * @param messagePO
     * @param streamPO  消息的内容
     */
    private void sendToMq(String topicKey, String topicFlag, MsgPO messagePO, MsgStreamPO streamPO) {
        MsgPO po = new MsgPO();
        po.setMsgId(messagePO.getMsgId());
        po.setAppKey(messagePO.getAppKey());
        po.setTitle(messagePO.getTitle());
        po.setBody(messagePO.getBody());
        po.setExtField1(messagePO.getAppId().toString());
        po.setExtField2(streamPO.getTargetId());
        po.setExtField3(streamPO.getStreamId().toString());
//        String content = JsonUtil.pojoToJson(streamPO);
        String content = JsonUtil.pojoToJson(po);

        try {

            Message message = new Message(topicKey /* Topic */,
                    topicFlag /* Tag */,
                    String.valueOf(streamPO.getMsgId()),
                    (content).getBytes(RemotingHelper.DEFAULT_CHARSET));

            SendResult result = messageRelayService.send(message);
            log.debug("result={}",result);
        } catch (Exception e) {
            e.printStackTrace();
            throw BusinessException.builder().errMsg("构造消息出错").build();
        }

    }


}
